# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: site-packages\urllib3\connection.py
from __future__ import absolute_import
import datetime, logging, os, sys, socket
from socket import error as SocketError, timeout as SocketTimeout
import warnings
from .packages import six
from .packages.six.moves.http_client import HTTPConnection as _HTTPConnection
from .packages.six.moves.http_client import HTTPException

try:
    import ssl

    BaseSSLError = ssl.SSLError
except (ImportError, AttributeError):
    ssl = None

    class BaseSSLError(BaseException):
        pass


try:
    ConnectionError = ConnectionError
except NameError:

    class ConnectionError(Exception):
        pass


from .exceptions import (
    NewConnectionError,
    ConnectTimeoutError,
    SubjectAltNameWarning,
    SystemTimeWarning,
)
from .packages.ssl_match_hostname import match_hostname, CertificateError
from .util.ssl_ import (
    resolve_cert_reqs,
    resolve_ssl_version,
    assert_fingerprint,
    create_urllib3_context,
    ssl_wrap_socket,
)
from .util import connection
from ._collections import HTTPHeaderDict

log = logging.getLogger(__name__)
port_by_scheme = {"http": 80, "https": 443}
RECENT_DATE = datetime.date(2016, 1, 1)


class DummyConnection(object):
    """Used to detect a failed ConnectionCls import."""

    pass


class HTTPConnection(_HTTPConnection, object):
    """
    Based on httplib.HTTPConnection but provides an extra constructor
    backwards-compatibility layer between older and newer Pythons.
    
    Additional keyword parameters are used to configure attributes of the connection.
    Accepted parameters include:
    
      - ``strict``: See the documentation on :class:`urllib3.connectionpool.HTTPConnectionPool`
      - ``source_address``: Set the source address for the current connection.
    
        .. note:: This is ignored for Python 2.6. It is only applied for 2.7 and 3.x
    
      - ``socket_options``: Set specific options on the underlying socket. If not specified, then
        defaults are loaded from ``HTTPConnection.default_socket_options`` which includes disabling
        Nagle's algorithm (sets TCP_NODELAY to 1) unless the connection is behind a proxy.
    
        For example, if you wish to enable TCP Keep Alive in addition to the defaults,
        you might pass::
    
            HTTPConnection.default_socket_options + [
                (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
            ]
    
        Or you may want to disable the defaults by passing an empty list (e.g., ``[]``).
    """

    default_port = port_by_scheme["http"]
    default_socket_options = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
    is_verified = False

    def __init__(self, *args, **kw):
        if six.PY3:
            kw.pop("strict", None)
        self.source_address = kw.get("source_address")
        if sys.version_info < (2, 7):
            kw.pop("source_address", None)
        self.socket_options = kw.pop("socket_options", self.default_socket_options)
        _HTTPConnection.__init__(self, *args, **kw)

    def _new_conn(self):
        """ Establish a socket connection and set nodelay settings on it.
        
        :return: New socket connection.
        """
        extra_kw = {}
        if self.source_address:
            extra_kw["source_address"] = self.source_address
        if self.socket_options:
            extra_kw["socket_options"] = self.socket_options
        try:
            conn = connection.create_connection(
                (self.host, self.port), self.timeout, **extra_kw
            )
        except SocketTimeout as e:
            raise ConnectTimeoutError(
                self,
                "Connection to %s timed out. (connect timeout=%s)"
                % (self.host, self.timeout),
            )
        except SocketError as e:
            raise NewConnectionError(
                self, "Failed to establish a new connection: %s" % e
            )

        return conn

    def _prepare_conn(self, conn):
        self.sock = conn
        if getattr(self, "_tunnel_host", None):
            self._tunnel()
            self.auto_open = 0

    def connect(self):
        conn = self._new_conn()
        self._prepare_conn(conn)

    def request_chunked(self, method, url, body=None, headers=None):
        """
        Alternative to the common request method, which sends the
        body with chunked encoding and not as one block
        """
        headers = HTTPHeaderDict(headers if headers is not None else {})
        skip_accept_encoding = "accept-encoding" in headers
        skip_host = "host" in headers
        self.putrequest(
            method, url, skip_accept_encoding=skip_accept_encoding, skip_host=skip_host
        )
        for header, value in headers.items():
            self.putheader(header, value)

        if "transfer-encoding" not in headers:
            self.putheader("Transfer-Encoding", "chunked")
        self.endheaders()
        if body is not None:
            stringish_types = six.string_types + (six.binary_type,)
            if isinstance(body, stringish_types):
                body = (body,)
            for chunk in body:
                if not chunk:
                    continue
                if not isinstance(chunk, six.binary_type):
                    chunk = chunk.encode("utf8")
                len_str = hex(len(chunk))[2:]
                self.send(len_str.encode("utf-8"))
                self.send(b"\r\n")
                self.send(chunk)
                self.send(b"\r\n")

        self.send(b"0\r\n\r\n")


class HTTPSConnection(HTTPConnection):
    default_port = port_by_scheme["https"]
    ssl_version = None

    def __init__(
        self,
        host,
        port=None,
        key_file=None,
        cert_file=None,
        strict=None,
        timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
        ssl_context=None,
        **kw
    ):
        HTTPConnection.__init__(strict=strict, timeout=timeout, **kw)
        self.key_file = key_file
        self.cert_file = cert_file
        self.ssl_context = ssl_context
        self._protocol = "https"

    def connect(self):
        conn = self._new_conn()
        self._prepare_conn(conn)
        if self.ssl_context is None:
            self.ssl_context = create_urllib3_context(
                ssl_version=resolve_ssl_version(None), cert_reqs=resolve_cert_reqs(None)
            )
        self.sock = ssl_wrap_socket(
            sock=conn,
            keyfile=self.key_file,
            certfile=self.cert_file,
            ssl_context=self.ssl_context,
        )


class VerifiedHTTPSConnection(HTTPSConnection):
    """
    Based on httplib.HTTPSConnection but wraps the socket with
    SSL certification.
    """

    cert_reqs = None
    ca_certs = None
    ca_cert_dir = None
    ssl_version = None
    assert_fingerprint = None

    def set_cert(
        self,
        key_file=None,
        cert_file=None,
        cert_reqs=None,
        ca_certs=None,
        assert_hostname=None,
        assert_fingerprint=None,
        ca_cert_dir=None,
    ):
        """
        This method should only be called once, before the connection is used.
        """
        if cert_reqs is None:
            if ca_certs or ca_cert_dir:
                cert_reqs = "CERT_REQUIRED"
            else:
                if self.ssl_context is not None:
                    cert_reqs = self.ssl_context.verify_mode
            self.key_file = key_file
            self.cert_file = cert_file
            self.cert_reqs = cert_reqs
            self.assert_hostname = assert_hostname
            self.assert_fingerprint = assert_fingerprint
            self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
            self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)

    def connect(self):
        conn = self._new_conn()
        hostname = self.host
        if getattr(self, "_tunnel_host", None):
            self.sock = conn
            self._tunnel()
            self.auto_open = 0
            hostname = self._tunnel_host
        is_time_off = datetime.date.today() < RECENT_DATE
        if is_time_off:
            warnings.warn(
                (
                    "System time is way off (before {0}). This will probably lead to SSL verification errors"
                ).format(RECENT_DATE),
                SystemTimeWarning,
            )
        if self.ssl_context is None:
            self.ssl_context = create_urllib3_context(
                ssl_version=resolve_ssl_version(self.ssl_version),
                cert_reqs=resolve_cert_reqs(self.cert_reqs),
            )
        context = self.ssl_context
        context.verify_mode = resolve_cert_reqs(self.cert_reqs)
        self.sock = ssl_wrap_socket(
            sock=conn,
            keyfile=self.key_file,
            certfile=self.cert_file,
            ca_certs=self.ca_certs,
            ca_cert_dir=self.ca_cert_dir,
            server_hostname=hostname,
            ssl_context=context,
        )
        if self.assert_fingerprint:
            assert_fingerprint(
                self.sock.getpeercert(binary_form=True), self.assert_fingerprint
            )
        else:
            if (
                context.verify_mode != ssl.CERT_NONE
                and not getattr(context, "check_hostname", False)
                and self.assert_hostname is not False
            ):
                cert = self.sock.getpeercert()
                if not cert.get("subjectAltName", ()):
                    warnings.warn(
                        (
                            "Certificate for {0} has no `subjectAltName`, falling back to check for a `commonName` for now. This feature is being removed by major browsers and deprecated by RFC 2818. (See https://github.com/shazow/urllib3/issues/497 for details.)"
                        ).format(hostname),
                        SubjectAltNameWarning,
                    )
                _match_hostname(cert, self.assert_hostname or hostname)
        self.is_verified = (
            context.verify_mode == ssl.CERT_REQUIRED
            or self.assert_fingerprint is not None
        )


def _match_hostname(cert, asserted_hostname):
    try:
        match_hostname(cert, asserted_hostname)
    except CertificateError as e:
        log.error(
            "Certificate did not match expected hostname: %s. Certificate: %s",
            asserted_hostname,
            cert,
        )
        e._peer_cert = cert
        raise


if ssl:
    UnverifiedHTTPSConnection = HTTPSConnection
    HTTPSConnection = VerifiedHTTPSConnection
else:
    HTTPSConnection = DummyConnection
